package com.wordcount;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * @author yuanzheng
 * @date 2020/4/26-22:08
 */
public class WordCountJob extends Configured implements Tool {

    public static void main(String[] args) throws Exception {
        //执行job作业任务类的对象是谁
        ToolRunner.run(new WordCountJob(), args);
    }

    /**
     * job程序
     *
     * @param strings 参数
     * @return 0:1
     * @throws Exception
     */
    @Override
    public int run(String[] strings) throws Exception {
        //创建job作业对象
        Job job = Job.getInstance();
        job.setJarByClass(WordCountJob.class);

        // 1.inputformat
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path("/home/hadoop/yuan.log"));

        //支持设置一个hdfs系统目录，job作业会在执行时将这个目录中所有的文件都读取
        //TextInputFormat.addInputPath(job, new Path("/wordcount/"));

        // 2.map
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 3.shuffle 自动执行

        // 4.reduce
        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 5.outputformat
        job.setOutputFormatClass(TextOutputFormat.class);
        Path res = new Path("/home/hadoop/result/");
        FileSystem fileSystem = FileSystem.get(getConf());
        if (fileSystem.exists(res)) {
            fileSystem.delete(res, true);
        }
        TextOutputFormat.setOutputPath(job, res);
        //保证输出结果的目录不存在

        // 6.commit job
        //job.submit();
        boolean status = job.waitForCompletion(true);
        System.out.println("word count status : " + status);
        return job.waitForCompletion(true) ? 0 : 1;
    }

}
